package com.xuanyuan.atlas.util;

import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author LUOYUAN
 * @create 2020-05-14-12:02
 */
public class FluxFeignUtil {



    public static Builder feignBuilder(){
        return new Builder();
    }



    public static class Builder{
        private ReentrantLock lock = new ReentrantLock();
        private Condition condition = lock.newCondition();
        private AtomicInteger num = new AtomicInteger(0);
        private List<Exec> execs = new ArrayList<>();

        public Builder add(Exec fun){
            num.incrementAndGet();
            execs.add(fun);
            return this;
        }

        public void block() throws InterruptedException {
            lock.lock();
            try{
                execs.forEach(exec -> {
                    Mono mono = exec.feign();
                    mono.subscribe(o -> {
                        try{
                            exec.subscribe(o);
                        }catch (Exception e){
                            e.printStackTrace();
                        }finally {
                            cancellation();
                        }
                    });
                });
                if(num.get()!=0){
                    condition.await(10000, TimeUnit.SECONDS);
                }
            }finally {
                lock.unlock();
            }
            System.out.println("执行结束");
        }

        private void cancellation(){
            lock.lock();
            try{
                if(num.decrementAndGet()==0){
                    condition.signal();
                }
            }finally {
                lock.unlock();
            }
        }
    }



    public  interface Exec<T>{
        Mono<T> feign();
        void subscribe(T t);
    }

}